package com.jsean.demo;

import lombok.extern.slf4j.Slf4j;
import org.apache.avro.file.DataFileReader;
import org.apache.avro.file.DataFileWriter;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificDatumWriter;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

@Slf4j
public class ParquetExample {

    private static String getParquetName() {
        Properties props = new Properties();
        try (InputStream is = ParquetExample.class.getClassLoader().getResourceAsStream("application.properties")) {
            props.load(is);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return props.getProperty("app.parquet.file", "users.parquet");
    }

    public static void main(String[] args) throws IOException {
        String filePath = getParquetName();

        // ✅ Create: 写入数据到 Parquet 文件
        writeParquet(filePath);

        // 🔍 Read: 读取 Parquet 文件数据
        List<User> users = readParquet(filePath);
        System.out.println("Read Users:");
        users.forEach(System.out::println);

        // 🔄 Update: 修改其中一个用户的 email
        for (User user : users) {
            if (user.getId() == 1) {
                user.setEmail("new.email@example.com");
            }
        }

        // 🗑️ Delete: 删除 id=2 的用户
        List<User> updatedUsers = new ArrayList<>();
        for (User user : users) {
            if (user.getId() != 2) {
                updatedUsers.add(user);
            }
        }

        // 重新写入更新后的数据
        String updatedFilePath = "updated_users.parquet";
        writeParquet(updatedFilePath, updatedUsers);

        // 再次读取查看是否更新成功
        System.out.println("\nUpdated Users:");
        readParquet(updatedFilePath).forEach(System.out::println);
    }

    // 写入 Parquet 文件
    private static void writeParquet(String path) throws IOException {
        List<User> users = new ArrayList<>();
        users.add(User.newBuilder().setId(1).setName("Alice").setEmail("alice@example.com").build());
        users.add(User.newBuilder().setId(2).setName("Bob").build());
        users.add(User.newBuilder().setId(3).setName("Charlie").setEmail("charlie@example.com").build());

        writeParquet(path, users);
    }

    private static void writeParquet(String path, List<User> users) throws IOException {
        SpecificDatumWriter<User> datumWriter = new SpecificDatumWriter<>(User.class);
        DataFileWriter<User> dataFileWriter = new DataFileWriter<>(datumWriter);
        dataFileWriter.create(users.get(0).getSchema(), new File(path));

        for (User user : users) {
            dataFileWriter.append(user);
        }

        dataFileWriter.close();
        System.out.println("Written to " + path);
    }

    // 读取 Parquet 文件
    private static List<User> readParquet(String path) throws IOException {
        List<User> result = new ArrayList<>();

        SpecificDatumReader<User> datumReader = new SpecificDatumReader<>(User.class);
        DataFileReader<User> dataFileReader = new DataFileReader<>(new File(path), datumReader);

        User user;
        while (dataFileReader.hasNext()) {
            user = dataFileReader.next();
            result.add(user.deepCopy()); // 避免引用问题
        }

        dataFileReader.close();
        return result;
    }
}